package com.ecommerce.function;
import com.ecommerce.model.SearchWordData;import org.apache.flink.api.common.state.ListState;import org.apache.flink.api.common.state.ListStateDescriptor;import org.apache.flink.configuration.Configuration;import org.apache.flink.streaming.api.functions.KeyedProcessFunction;import org.apache.flink.util.Collector;
import java.util.ArrayList;import java.util.Comparator;import java.util.List;
public class SearchRankingFunction extends KeyedProcessFunction<String, SearchWordData, SearchWordData> {
    
    private final String rankType;
    private transient ListState<SearchWordData> searchState;
    
    public SearchRankingFunction(String rankType) {
        this.rankType = rankType;
    }
    
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        ListStateDescriptor<SearchWordData> descriptor = 
            new ListStateDescriptor<>("searchState", SearchWordData.class);
        searchState = getRuntimeContext().getListState(descriptor);
    }
    
    @Override
    public void processElement(SearchWordData word, Context ctx, Collector<SearchWordData> out) throws Exception {
        searchState.add(word);
        long triggerTime = (System.currentTimeMillis() / 1800000 + 1) * 1800000; // 30分钟
        ctx.timerService().registerProcessingTimeTimer(triggerTime);
    }
    
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<SearchWordData> out) throws Exception {
        List<SearchWordData> words = new ArrayList<>();
        for (SearchWordData word : searchState.get()) {
            words.add(word);
        }
        
        // 根据排名类型排序
        if ("popularity".equals(rankType)) {
            words.sort(Comparator.comparing(SearchWordData::getSearchCount).reversed());
        } else if ("growth".equals(rankType)) {
            words.sort(Comparator.comparing(SearchWordData::getGrowthRate).reversed());
        } else if ("blue_ocean".equals(rankType)) {
            words.sort(Comparator.comparing(SearchWordData::getSupplyDemandRatio).reversed());
        }
        
        int rank = 1;
        for (int i = 0; i < Math.min(100, words.size()); i++) {
            SearchWordData word = words.get(i);
            word.setRank(rank);
            word.setRankType(rankType);
            word.setTimestamp(System.currentTimeMillis());
            out.collect(word);
            rank++;
        }
        
        searchState.clear();
    }}